package data.platform.rest;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import data.platform.common.service.command.RuleScheduleService;
import data.platform.common.service.query.RulePointQueryService;
import data.platform.common.service.query.RuleQueryService;
import data.platform.common.service.query.WindowAggregateQueryService;
import data.platform.common.stream.aggregates.WindowAggregateItemVO;
import data.platform.common.stream.aggregates.WindowResultVO;
import data.platform.common.util.DateUtil;
import data.platform.rest.domain.RulePointQuery;
import data.platform.rest.domain.RulePointResultVO;
import data.platform.rest.domain.RulePointVO;
import data.platform.rest.domain.RuleVO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@RestController
@RequestMapping("/api/v1/stream")
@Slf4j
@RequiredArgsConstructor
public class StreamController {

    final RuleQueryService ruleQueryService;

    final WindowAggregateQueryService windowAggregateQueryService;

    final RulePointQueryService rulePointQueryService;

    final RuleScheduleService ruleScheduleService;

    @GetMapping(value = "/rule")
    public Mono<ResponseEntity<List<RuleVO>>> ruleQuery() {
        return ruleQueryService.findAll()
                .map(rule -> RuleVO.toRuleVO(rule))
                .collectList()
                .elapsed()
                .map(tuple2 -> ResponseEntity.ok()
                        .header("executetime", tuple2.getT1().toString())
                        .body(tuple2.getT2()));
    }

    @PostMapping(value = "/rule")
    public Mono<ResponseEntity<RuleVO>> createRule(@RequestBody RuleVO ruleVO) {
        return Mono.just(ruleVO)
                .map(vo -> vo.toRule())
                .flatMap(rule -> ruleScheduleService.createOrUpdateRule(rule))
                .map(rule -> RuleVO.toRuleVO(rule))
                .elapsed()
                .map(tuple2 -> ResponseEntity.ok()
                        .header("executetime", tuple2.getT1().toString())
                        .body(tuple2.getT2()));
    }

    @GetMapping(value = "/rule/{ruleId}")
    public Mono<ResponseEntity<RuleVO>> findRule(@PathVariable String ruleId) {
        return ruleQueryService.findById(ruleId)
                .map(rule -> RuleVO.toRuleVO(rule))
                .elapsed()
                .map(tuple2 -> ResponseEntity.ok()
                        .header("executetime", tuple2.getT1().toString())
                        .body(tuple2.getT2()));
    }

    @PutMapping(value = "/rule/{ruleState}/{ruleId}")
    public Mono<ResponseEntity<String>> findRule(@PathVariable String ruleState, @PathVariable String ruleId) {
        if (ruleState.equals("ACTIVE")) {
            return ruleScheduleService.startRule(ruleId)
                    .elapsed()
                    .map(tuple2 -> ResponseEntity.ok()
                            .header("executetime", tuple2.getT1().toString())
                            .body(tuple2.getT2()));
        } else {
            return ruleScheduleService.pauseRule(ruleId)
                    .elapsed()
                    .map(tuple2 -> ResponseEntity.ok()
                            .header("executetime", tuple2.getT1().toString())
                            .body(tuple2.getT2()));
        }
    }

    @GetMapping(value = "/windowAggregate/{ruleId}")
    public Mono<ResponseEntity<List<WindowAggregateItemVO>>> windowAggregateQuery(@PathVariable String ruleId) {
        return windowAggregateQueryService.queryAggregateItems(ruleId)
                .collectList()
                .elapsed()
                .map(tuple2 -> ResponseEntity.ok()
                        .header("executetime", tuple2.getT1().toString())
                        .body(tuple2.getT2()));
    }

    @GetMapping(value = "/windowResult/{ruleId}")
    public Mono<ResponseEntity<List<WindowResultVO>>> windowResultQuery(@PathVariable String ruleId) {
        return windowAggregateQueryService.queryWindowResult(ruleId)
                .collectList()
                .elapsed()
                .map(tuple2 -> ResponseEntity.ok()
                        .header("executetime", tuple2.getT1().toString())
                        .body(tuple2.getT2()));
    }

    @PostMapping(value = "/ruleResult")
    public Mono<ResponseEntity<List<RulePointVO>>> ruleResultQuery(@RequestBody RulePointQuery rulePointQuery) {
        ObjectMapper objectMapper = new ObjectMapper();
        return rulePointQueryService.query(DateUtil.getDateTimeOfTimestamp(rulePointQuery.getBeginTime()),
                DateUtil.getDateTimeOfTimestamp(rulePointQuery.getEndTime()), rulePointQuery.getRuleId())
                .map(rulePoint -> {
                    RulePointVO vo = new RulePointVO();
                    vo.setId(rulePoint.getRuleId());
                    vo.setEventTime(DateUtil.getDateOfLocalDateTime(rulePoint.getEventTime()).getTime());
                    vo.setHitNumber(rulePoint.getHitNumber());
                    try {
                        Map<String, Object> hitDataMap = objectMapper.readValue(rulePoint.getHitData(), Map.class);
                        vo.setStartTime((long) hitDataMap.get("startTime"));
                        vo.setEndTime((long) hitDataMap.get("endTime"));

                        List<RulePointResultVO> resultVOS = new ArrayList<>();
                        vo.setResultVOS(resultVOS);

                        Map resultMap = (Map) hitDataMap.get("result");
                        resultMap.forEach((tag, resultValueMap) -> {
                            Map windowResultMap = (Map) resultValueMap;
                            Map metricValueMap = (Map) windowResultMap.get("metricValue");
                            RulePointResultVO rulePointResultVO = new RulePointResultVO();
                            try {
                                rulePointResultVO.setTags(objectMapper.readValue(tag.toString(), Map.class));
                            } catch (JsonProcessingException e) {
                                log.error("", e);
                            }
                            rulePointResultVO.setMetricValues(metricValueMap);

                            resultVOS.add(rulePointResultVO);
                        });
                    } catch (JsonProcessingException e) {
                        log.error("", e);
                    }

                    return vo;
                })
                .collectList()
                .elapsed()
                .map(tuple2 -> ResponseEntity.ok()
                        .header("executetime", tuple2.getT1().toString())
                        .body(tuple2.getT2()));
    }
}
